package org.zjt.http;

import io.reactivex.Flowable;
import io.reactivex.Maybe;
import io.reactivex.Observable;
import io.reactivex.Single;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.util.EntityUtils;

import java.io.IOException;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.concurrent.CountDownLatch;

/**
 * Description:
 *
 * @author juntao.zhang
 * Date: 2018-05-29 下午5:17
 *
    类型	            描述
    Observable<T>	能够发射0或n个数据，并以成功或错误事件终止。
    Flowable<T>	    能够发射0或n个数据，并以成功或错误事件终止。 支持Backpressure（背压），可以控制数据源发射的速度。
    Single<T>	    只发射单个数据或错误事件。
    Completable	    它从来不发射数据，只处理 onComplete 和 onError 事件。可以看成是Rx的Runnable。
    Maybe<T>	    能够发射0或者1个数据，要么成功，要么失败。有点类似于Optional

    从上面的表格可以看出，这五种被观察者类型中只有Flowable能支持Backpressure，如果有需要Backpressure的情况，还是必须要使用Flowable。

作者：fengzhizi715
链接：https://www.jianshu.com/p/45309538ad94

 */
@Slf4j
public class ReactiveAsyncHttpTest {


    public static void main(String[] args) throws InterruptedException {

        Flowable.range(1, 10).all(v -> v > 0).subscribe(v -> log.info(String.valueOf(v)));


        Single.create(source -> source.onSuccess("zhang")).map(v -> v.toString().toUpperCase()).subscribe(v -> log.info(String.valueOf(v)));


        // ParallelFlowable 使用 sequential + blockingSubscribe 顺序化
        Flowable.range(1, 10).parallel().runOn(Schedulers.computation()).map(v -> v * v).sequential().blockingSubscribe(System.out::println);




        Maybe.<Integer>create(emitter -> {
            emitter.onSuccess(222);
            emitter.onComplete();

            emitter.setCancellable(() -> log.info("-----cancel取消------"));
            emitter.setDisposable(new Disposable() {
                @Override
                public void dispose() {
                    log.info("-----dispose------");
                }

                @Override
                public boolean isDisposed() {
                    log.info("-----isDisposed------");
                    return false;
                }
            });
        }).subscribe(v -> log.info(String.valueOf(v)));


    }


    /**
     * 单http reactive 请求
     */
    public static void rxHttpSingleExec() {
        RequestConfig requestConfig = RequestConfig.custom().setConnectTimeout(2000).setSocketTimeout(2000).setConnectionRequestTimeout(2000).build();

        try (final CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom().setMaxConnTotal(100).setMaxConnPerRoute(30)
                .setDefaultRequestConfig(requestConfig)
                .build()) {

            Observable.create(emitter ->{

                httpclient.start();

                final HttpGet request2 = new HttpGet("https://www.baidu.com/");

                httpclient.execute(request2, new FutureCallback<HttpResponse>() {
                    @Override
                    public void completed(final HttpResponse response2) {

                        try {
                            String string = EntityUtils.toString(response2.getEntity());
                            emitter.onNext(string);
                            emitter.onComplete();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                        log.info(request2.getRequestLine() + "->" + response2.getStatusLine());
                    }

                    @Override
                    public void failed(final Exception ex) {
                        log.error(request2.getRequestLine() + "->" + ex);
                        emitter.onError(ex);
                        emitter.onComplete();
                    }

                    @Override
                    public void cancelled() {
                        emitter.onNext(null);
                        emitter.onComplete();
                        log.info(request2.getRequestLine() + " cancelled");
                    }
                });
            }).subscribeOn(Schedulers.io())
                    .observeOn(Schedulers.single())
                    .map(String::valueOf)
                    .subscribe(log::info);


            Thread.sleep(50000);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 多http 请求
     */
    public static void rxHttpMutilExec() {

        RequestConfig requestConfig = RequestConfig.custom().setConnectTimeout(5000).setSocketTimeout(5000).setConnectionRequestTimeout(5000).build();

        try (final CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom().setMaxConnTotal(1000).setMaxConnPerRoute(300)
                .setDefaultRequestConfig(requestConfig)
                .build()) {

            CountDownLatch countDownLatch = new CountDownLatch(1000);

            // Start the client
            new Thread(()-> Observable.range(1, 1000).flatMap(v -> Observable.create(emitter -> {
                httpclient.start();
                final HttpGet request2 = new HttpGet("https://www.baidu.com/");
                httpclient.execute(request2, new FutureCallback<HttpResponse>() {
                    @Override
                    public void completed(final HttpResponse response2) {
                        try {
                            countDownLatch.countDown();
                            String string = EntityUtils.toString(response2.getEntity());
                            emitter.onNext(string);
                            emitter.onComplete();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                        log.info(request2.getRequestLine() + "->" + response2.getStatusLine());
                    }

                    @Override
                    public void failed(final Exception ex) {
                        log.error(request2.getRequestLine() + "->" + ex);
                        emitter.onError(ex);
                        emitter.onComplete();
                        countDownLatch.countDown();
                    }

                    @Override
                    public void cancelled() {
                        emitter.onNext(null);
                        emitter.onComplete();
                        log.info(request2.getRequestLine() + " cancelled");
                        countDownLatch.countDown();
                    }
                });

            }).subscribeOn(Schedulers.io()).observeOn(Schedulers.single()).map(String::valueOf)).subscribe(log::info)).start();

            countDownLatch.await();
            log.info("countDownLatch:{}",countDownLatch.getCount() , LocalDateTime.now().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
